package com.leon.datalink.rule.persistence;

import akka.persistence.AbstractPersistentActor;
import akka.persistence.SnapshotOffer;
import com.leon.datalink.core.persistence.PersistenceConfigManager;
import com.leon.datalink.core.persistence.NodeActorStateAnalysis;
import com.leon.datalink.core.persistence.SnapshotStrategyManager;
import com.leon.datalink.core.persistence.PersistenceMetricsManager;
import com.leon.datalink.core.persistence.event.NodeEvent;
import com.leon.datalink.core.persistence.event.NodeStartedEvent;
import com.leon.datalink.core.persistence.event.NodeConfiguredEvent;
import com.leon.datalink.core.persistence.event.DataProcessedEvent;
import com.leon.datalink.core.persistence.state.NodeState;
import com.leon.datalink.core.persistence.state.NodeStatus;
import com.leon.datalink.node.AbstractNodeActor;
import com.leon.datalink.node.NodeActorCreateParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 持久化Actor包装器
 * 为现有Actor提供透明的持久化支持
 * 
 * @author leon
 */
public class PersistentActorWrapper extends AbstractPersistentActor {
    
    private static final Logger logger = LoggerFactory.getLogger(PersistentActorWrapper.class);
    
    private final Class<? extends AbstractNodeActor> wrappedActorClass;
    private final NodeActorCreateParam createParam;
    private AbstractNodeActor wrappedActor;
    private NodeState currentState;

    // 动态快照频率配置（基于Actor类型）
    private final int snapshotFrequency;
    private final int maxSnapshots;
    private int eventCount = 0;
    private final int actorPriority;

    // 智能快照策略管理器
    private final SnapshotStrategyManager snapshotManager;

    // 性能监控管理器
    private final PersistenceMetricsManager metricsManager;

    // 性能监控指标
    private long totalPersistTime = 0;
    private long totalRecoveryTime = 0;
    private int persistCount = 0;
    private int recoveryCount = 0;
    private long lastPersistTime = 0;
    
    public PersistentActorWrapper(Class<? extends AbstractNodeActor> wrappedActorClass,
                                 NodeActorCreateParam createParam) {
        this.wrappedActorClass = wrappedActorClass;
        this.createParam = createParam;
        this.currentState = new NodeState();
        this.currentState.setNodeId(createParam.getNodeId());
        this.currentState.setStatus(NodeStatus.INITIALIZING);
        this.currentState.setProcessedCount(0L);
        this.currentState.setLastProcessedTime(System.currentTimeMillis());

        // 根据Actor类型设置持久化策略
        this.actorPriority = NodeActorStateAnalysis.getActorPersistencePriority(wrappedActorClass.getName());
        this.snapshotFrequency = getSnapshotFrequencyByPriority(actorPriority);
        this.maxSnapshots = getMaxSnapshotsByPriority(actorPriority);

        // 初始化智能快照策略管理器和性能监控管理器
        this.snapshotManager = SnapshotStrategyManager.getInstance();
        this.metricsManager = PersistenceMetricsManager.getInstance();

        logger.info("Created persistent wrapper for {}, priority={}, snapshotFreq={}, maxSnapshots={}",
                   wrappedActorClass.getSimpleName(), actorPriority, snapshotFrequency, maxSnapshots);
    }
    
    @Override
    public String persistenceId() {
        return "node-" + createParam.getNodeId();
    }
    
    @Override
    public void preStart() throws Exception {
        super.preStart();
        logger.info("Starting persistent actor wrapper for node: {}", createParam.getNodeId());
        
        // 创建被包装的Actor实例
        createWrappedActor();
        
        // 持久化启动事件
        NodeStartedEvent startEvent = new NodeStartedEvent();
        startEvent.setNodeId(createParam.getNodeId());
        startEvent.setTimestamp(System.currentTimeMillis());
        startEvent.setActorClass(wrappedActorClass.getName());
        
        long startTime = System.currentTimeMillis();
        persist(startEvent, event -> {
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            totalPersistTime += duration;
            persistCount++;
            lastPersistTime = endTime;

            // 记录性能指标
            metricsManager.recordPersistOperation(wrappedActorClass.getSimpleName(),
                                                createParam.getNodeId(),
                                                duration,
                                                estimateEventSize(event));

            updateState(event);
            logger.debug("Persisted start event for node: {} in {}ms", createParam.getNodeId(), duration);
        });
    }
    
    @Override
    public void postStop() throws Exception {
        if (wrappedActor != null) {
            try {
                wrappedActor.postStop();
            } catch (Exception e) {
                logger.error("Error stopping wrapped actor for node: {}", createParam.getNodeId(), e);
            }
        }
        super.postStop();
        logger.info("Stopped persistent actor wrapper for node: {}", createParam.getNodeId());
    }
    
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(String.class, cmd -> {
                    if ("recover".equals(cmd)) {
                        // 恢复逻辑
                        createReceiveRecover().onMessage().apply(cmd);
                    } else {
                        // 命令处理逻辑
                        createReceiveCommand().onMessage().apply(cmd);
                    }
                })
                .matchAny(message -> {
                    // 默认使用命令处理
                    createReceiveCommand().onMessage().apply(message);
                })
                .build();
    }

    public Receive createReceiveRecover() {
        return receiveBuilder()
                .match(NodeEvent.class, event -> {
                    long startTime = System.currentTimeMillis();
                    updateState(event);
                    long endTime = System.currentTimeMillis();
                    long duration = endTime - startTime;
                    totalRecoveryTime += duration;
                    recoveryCount++;

                    // 记录恢复性能指标
                    metricsManager.recordRecoveryOperation(wrappedActorClass.getSimpleName(),
                                                         createParam.getNodeId(),
                                                         duration,
                                                         1); // 单个事件
                })
                .match(SnapshotOffer.class, snapshot -> {
                    if (snapshot.snapshot() instanceof NodeState) {
                        long startTime = System.currentTimeMillis();
                        currentState = (NodeState) snapshot.snapshot();
                        long endTime = System.currentTimeMillis();
                        long duration = endTime - startTime;
                        totalRecoveryTime += duration;
                        recoveryCount++;

                        // 记录快照恢复性能指标
                        metricsManager.recordRecoveryOperation(wrappedActorClass.getSimpleName(),
                                                             createParam.getNodeId(),
                                                             duration,
                                                             1); // 快照恢复

                        logger.info("Recovered state from snapshot for node: {} in {}ms, processed count: {}",
                                  currentState.getNodeId(), duration, currentState.getProcessedCount());
                    }
                })
                .build();
    }
    
    public Receive createReceiveCommand() {
        return receiveBuilder()
                .matchAny(message -> {
                    // 将消息转发给被包装的Actor
                    if (wrappedActor != null) {
                        try {
                            // 记录数据处理事件
                            if (shouldPersistMessage(message)) {
                                DataProcessedEvent processEvent = new DataProcessedEvent();
                                processEvent.setNodeId(createParam.getNodeId());
                                processEvent.setTimestamp(System.currentTimeMillis());
                                processEvent.setMessageType(message.getClass().getSimpleName());
                                processEvent.setDataSize(calculateMessageSize(message));
                                
                                long startTime = System.currentTimeMillis();
                                persist(processEvent, event -> {
                                    long endTime = System.currentTimeMillis();
                                    long duration = endTime - startTime;
                                    totalPersistTime += duration;
                                    persistCount++;
                                    lastPersistTime = endTime;

                                    // 记录性能指标
                                    metricsManager.recordPersistOperation(wrappedActorClass.getSimpleName(),
                                                                        createParam.getNodeId(),
                                                                        duration,
                                                                        estimateEventSize(event));

                                    updateState(event);
                                    checkSnapshot();
                                });
                            }
                            
                            // 转发消息给被包装的Actor的receive方法
                            wrappedActor.createReceive().onMessage().apply(message);
                            
                        } catch (Exception e) {
                            logger.error("Error processing message in wrapped actor for node: {}", 
                                       createParam.getNodeId(), e);
                            getSender().tell(new akka.actor.Status.Failure(e), getSelf());
                        }
                    } else {
                        logger.warn("Wrapped actor not initialized for node: {}", createParam.getNodeId());
                        getSender().tell(new akka.actor.Status.Failure(
                            new IllegalStateException("Wrapped actor not initialized")), getSelf());
                    }
                })
                .build();
    }
    
    /**
     * 创建被包装的Actor实例
     */
    private void createWrappedActor() throws Exception {
        wrappedActor = wrappedActorClass.getDeclaredConstructor().newInstance();
        wrappedActor.setCreateParam(createParam);
        
        // 如果有恢复的状态，应用到被包装的Actor
        if (currentState.getStatus() != NodeStatus.INITIALIZING) {
            applyRecoveredState();
        }
        
        // 启动被包装的Actor
        wrappedActor.preStart();
        
        // 持久化配置事件
        NodeConfiguredEvent configEvent = new NodeConfiguredEvent();
        configEvent.setNodeId(createParam.getNodeId());
        configEvent.setTimestamp(System.currentTimeMillis());
        configEvent.setProperties(createParam.getProperties());
        
        long startTime = System.currentTimeMillis();
        persist(configEvent, event -> {
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            totalPersistTime += duration;
            persistCount++;
            lastPersistTime = endTime;

            // 记录性能指标
            metricsManager.recordPersistOperation(wrappedActorClass.getSimpleName(),
                                                createParam.getNodeId(),
                                                duration,
                                                estimateEventSize(event));

            updateState(event);
            currentState.setStatus(NodeStatus.RUNNING);
        });
    }
    
    /**
     * 应用恢复的状态到被包装的Actor
     */
    private void applyRecoveredState() {
        // 这里可以根据需要实现状态恢复逻辑
        logger.info("Applying recovered state to wrapped actor for node: {}, processed count: {}", 
                  currentState.getNodeId(), currentState.getProcessedCount());
    }
    
    /**
     * 更新内部状态
     */
    private void updateState(NodeEvent event) {
        // 记录状态变化前的状态
        NodeState oldState = cloneState(currentState);

        if (event instanceof NodeStartedEvent) {
            currentState.setStatus(NodeStatus.STARTING);
        } else if (event instanceof NodeConfiguredEvent) {
            NodeConfiguredEvent configEvent = (NodeConfiguredEvent) event;
            currentState.setProperties(configEvent.getProperties());
            currentState.setStatus(NodeStatus.RUNNING);
        } else if (event instanceof DataProcessedEvent) {
            DataProcessedEvent processEvent = (DataProcessedEvent) event;
            currentState.setProcessedCount(currentState.getProcessedCount() + 1);
            currentState.setLastProcessedTime(processEvent.getTimestamp());
            currentState.setLastMessageType(processEvent.getMessageType());
        }

        currentState.setLastEventTime(event.getTimestamp());

        // 记录状态变化到智能快照管理器
        snapshotManager.updateStateChangeStats(wrappedActorClass.getName(),
                                             createParam.getNodeId(),
                                             oldState,
                                             currentState);
    }

    /**
     * 克隆状态对象
     */
    private NodeState cloneState(NodeState state) {
        NodeState cloned = new NodeState();
        cloned.setNodeId(state.getNodeId());
        cloned.setStatus(state.getStatus());
        cloned.setProcessedCount(state.getProcessedCount());
        cloned.setLastProcessedTime(state.getLastProcessedTime());
        cloned.setLastMessageType(state.getLastMessageType());
        cloned.setLastEventTime(state.getLastEventTime());
        cloned.setProperties(state.getProperties());
        return cloned;
    }
    
    /**
     * 检查是否需要创建快照（使用智能快照策略）
     */
    private void checkSnapshot() {
        eventCount++;

        // 使用智能快照策略判断
        if (snapshotManager.shouldCreateSnapshot(wrappedActorClass.getName(),
                                               createParam.getNodeId(),
                                               eventCount,
                                               currentState)) {
            long startTime = System.currentTimeMillis();
            long snapshotSize = saveSnapshotWithSize(currentState);
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;

            // 记录快照性能指标
            snapshotManager.recordSnapshotCreated(wrappedActorClass.getName(),
                                                 createParam.getNodeId(),
                                                 snapshotSize);
            metricsManager.recordSnapshotOperation(wrappedActorClass.getSimpleName(),
                                                  createParam.getNodeId(),
                                                  duration,
                                                  snapshotSize);

            eventCount = 0;
            logger.debug("Saved intelligent snapshot for node: {} (priority: {}, size: {}KB, time: {}ms)",
                        createParam.getNodeId(), actorPriority, snapshotSize / 1024, duration);
        }
    }

    /**
     * 保存快照并返回大小
     */
    private long saveSnapshotWithSize(NodeState state) {
        saveSnapshot(state);
        // 简化实现：估算快照大小
        return estimateSnapshotSize(state);
    }

    /**
     * 估算快照大小
     */
    private long estimateSnapshotSize(NodeState state) {
        // 简化实现：基于状态内容估算大小
        long size = 1024; // 基础大小
        if (state.getProperties() != null) {
            size += state.getProperties().toString().length() * 2;
        }
        if (state.getLastMessageType() != null) {
            size += state.getLastMessageType().length() * 2;
        }
        return size;
    }

    /**
     * 估算事件大小
     */
    private long estimateEventSize(NodeEvent event) {
        // 简化实现：基于事件类型和内容估算大小
        long size = 512; // 基础大小

        if (event instanceof NodeConfiguredEvent) {
            NodeConfiguredEvent configEvent = (NodeConfiguredEvent) event;
            if (configEvent.getProperties() != null) {
                size += configEvent.getProperties().toString().length() * 2;
            }
        } else if (event instanceof DataProcessedEvent) {
            DataProcessedEvent processEvent = (DataProcessedEvent) event;
            if (processEvent.getMessageType() != null) {
                size += processEvent.getMessageType().length() * 2;
            }
        }

        return size;
    }

    /**
     * 根据Actor优先级获取快照频率
     */
    private int getSnapshotFrequencyByPriority(int priority) {
        switch (priority) {
            case 1: // 关键状态Actor
                return 100;
            case 2: // 业务状态Actor
                return 500;
            case 3: // 临时状态Actor
                return 1000;
            case 4: // 无状态Actor
            default:
                return 2000;
        }
    }

    /**
     * 根据Actor优先级获取最大快照数
     */
    private int getMaxSnapshotsByPriority(int priority) {
        switch (priority) {
            case 1: // 关键状态Actor
                return 5;
            case 2: // 业务状态Actor
                return 3;
            case 3: // 临时状态Actor
                return 2;
            case 4: // 无状态Actor
            default:
                return 1;
        }
    }
    
    /**
     * 判断是否需要持久化该消息
     */
    private boolean shouldPersistMessage(Object message) {
        // 根据配置决定是否持久化特定类型的消息
        return PersistenceConfigManager.getInstance().shouldPersistMessage(
            createParam.getNodeId(), message.getClass().getSimpleName());
    }
    
    /**
     * 计算消息大小（简化实现）
     */
    private long calculateMessageSize(Object message) {
        // 简化实现，实际可以使用序列化来计算准确大小
        return message.toString().length();
    }
    
    /**
     * 获取当前状态（用于监控）
     */
    public NodeState getCurrentState() {
        return currentState;
    }

    /**
     * 获取性能监控指标
     */
    public PersistenceMetrics getPerformanceMetrics() {
        return new PersistenceMetrics(
            createParam.getNodeId(),
            wrappedActorClass.getSimpleName(),
            actorPriority,
            persistCount,
            recoveryCount,
            totalPersistTime,
            totalRecoveryTime,
            persistCount > 0 ? totalPersistTime / persistCount : 0,
            recoveryCount > 0 ? totalRecoveryTime / recoveryCount : 0,
            eventCount,
            snapshotFrequency,
            maxSnapshots,
            System.currentTimeMillis() - lastPersistTime
        );
    }

    /**
     * 性能监控指标数据类
     */
    public static class PersistenceMetrics {
        private final String nodeId;
        private final String actorType;
        private final int priority;
        private final int persistCount;
        private final int recoveryCount;
        private final long totalPersistTime;
        private final long totalRecoveryTime;
        private final long avgPersistTime;
        private final long avgRecoveryTime;
        private final int currentEventCount;
        private final int snapshotFrequency;
        private final int maxSnapshots;
        private final long timeSinceLastPersist;

        public PersistenceMetrics(String nodeId, String actorType, int priority,
                                int persistCount, int recoveryCount,
                                long totalPersistTime, long totalRecoveryTime,
                                long avgPersistTime, long avgRecoveryTime,
                                int currentEventCount, int snapshotFrequency,
                                int maxSnapshots, long timeSinceLastPersist) {
            this.nodeId = nodeId;
            this.actorType = actorType;
            this.priority = priority;
            this.persistCount = persistCount;
            this.recoveryCount = recoveryCount;
            this.totalPersistTime = totalPersistTime;
            this.totalRecoveryTime = totalRecoveryTime;
            this.avgPersistTime = avgPersistTime;
            this.avgRecoveryTime = avgRecoveryTime;
            this.currentEventCount = currentEventCount;
            this.snapshotFrequency = snapshotFrequency;
            this.maxSnapshots = maxSnapshots;
            this.timeSinceLastPersist = timeSinceLastPersist;
        }

        // Getters
        public String getNodeId() { return nodeId; }
        public String getActorType() { return actorType; }
        public int getPriority() { return priority; }
        public int getPersistCount() { return persistCount; }
        public int getRecoveryCount() { return recoveryCount; }
        public long getTotalPersistTime() { return totalPersistTime; }
        public long getTotalRecoveryTime() { return totalRecoveryTime; }
        public long getAvgPersistTime() { return avgPersistTime; }
        public long getAvgRecoveryTime() { return avgRecoveryTime; }
        public int getCurrentEventCount() { return currentEventCount; }
        public int getSnapshotFrequency() { return snapshotFrequency; }
        public int getMaxSnapshots() { return maxSnapshots; }
        public long getTimeSinceLastPersist() { return timeSinceLastPersist; }

        @Override
        public String toString() {
            return String.format("PersistenceMetrics{nodeId='%s', actorType='%s', priority=%d, " +
                               "persistCount=%d, recoveryCount=%d, avgPersistTime=%dms, avgRecoveryTime=%dms, " +
                               "eventCount=%d/%d, timeSinceLastPersist=%dms}",
                               nodeId, actorType, priority, persistCount, recoveryCount,
                               avgPersistTime, avgRecoveryTime, currentEventCount, snapshotFrequency,
                               timeSinceLastPersist);
        }
    }
}
